package com.cdream.basic.mq.rocket;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * @author cdream
 * @date 2022/1/22
 */
public class DefaultMqExample {
    public static final String NAME_SRV = "cdream.zs:9876";

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //启动MQProducer
        // sendMessage();
        // receiveMessage();
        // sendMessageAsync();

    }

    private static void receiveMessage() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_start_consumer");
        consumer.setNamesrvAddr(NAME_SRV);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe("test_quick_topic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            MessageExt me = msgs.get(0);
            try {
                String body = new String(me.getBody(), RemotingHelper.DEFAULT_CHARSET);
                System.out.println("body = " + body);
            } catch (Exception e) {
                e.printStackTrace();
                me.getReconsumeTimes();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }

    private static void sendMessage() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("test_quick_start");
        producer.setNamesrvAddr(NAME_SRV);
        producer.start();
        // 消息发送
        Message message = new Message("test_quick_topic", "tags", "keys", "pig".getBytes());
        SendResult send = producer.send(message);
        System.out.println(send);
        producer.shutdown();
    }

    private static void sendMessageAsync() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("test_quick_start");
        producer.setNamesrvAddr(NAME_SRV);
        producer.start();
        // 消息发送
        Message message = new Message("test_quick_topic", "tags", "keys", "pig".getBytes());
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.println(e.getLocalizedMessage());
            }
        });
        // producer.shutdown();
    }

    private static void sendDelayMessage() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("test_quick_start");
        producer.setNamesrvAddr(NAME_SRV);
        producer.start();
        // 消息发送
        Message message = new Message("test_quick_topic", "tags", "keys", "pig".getBytes());
        // 固定精度
        message.setDelayTimeLevel(3);
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.println(e.getLocalizedMessage());
            }
        });
        // producer.shutdown();
    }

    private static void sendUserDefineMessage() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("test_quick_start");
        producer.setNamesrvAddr(NAME_SRV);
        producer.start();
        // 消息发送
        Message message = new Message("test_quick_topic", "tags", "keys", "pig".getBytes());
        // 固定精度
        message.setDelayTimeLevel(3);
        producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                int i = (int) arg;
                return mqs.get(i);
            }
        }, 1);
        // producer.shutdown();
    }
}
